<?php

namespace app; // 补充命名空间，避免全局类冲突


use app\command\klineMin;
use app\framework\lib\Log;
use app\framework\lib\Redis;
use app\service\CurrencyKline;
use Workerman\Connection\AsyncTcpConnection;
use Workerman\Crontab\Crontab;
use Workerman\Timer;
use Workerman\Worker;
use app\command\klineYesterDay;
use app\command\klineSecond;
use app\command\Order;
use InvalidArgumentException;
use Throwable;

/**
 * 控制台命令与数据处理类
 * 负责定时任务调度、火币数据订阅及行情数据组装
 */
class Console
{
    /**
     * 1秒K线的Redis键前缀
     */
    private const REDIS_KLINE_1SEC_PREFIX = 'market.%susdt.kline.1sec';

    /**
     * 昨日K线的Redis键前缀
     */
    private const REDIS_KLINE_YESTERDAY_PREFIX = 'market.%susdt.kline.yesterday';


    /**
     * 注册定时任务
     * 优化点：统一封装任务创建，增加错误捕获，确保进程名唯一
     */
    public static function command(): void
    {
        try {
            // 注册订单定时任务
            $orderTask = new Order('order', 60);
            self::createCrontabTask($orderTask);

            // 注册秒级K线定时任务
            $klineSecondTask = new klineSecond('kline-second', 60);
            self::createCrontabTask($klineSecondTask);

            // 注册昨日K线定时任务
            $klineYesterDayTask = new klineYesterDay('kline-yesterday', 30);
            self::createCrontabTask($klineYesterDayTask);
            $klineMinTask = new klineMin("kline-min",120);
            self::createCrontabTask($klineMinTask);
            Log::info("所有定时任务注册成功");
        } catch (Throwable $e) {
            Log::error("定时任务注册失败：{$e->getMessage()}", [
                'file' => $e->getFile(),
                'line' => $e->getLine()
            ]);
        }
    }

    /**
     * 创建定时任务（封装重复逻辑）
     */
    private static function createCrontabTask($task): void
    {
        if (empty($task->cron) || !is_callable([$task, 'run'])) {
            throw new InvalidArgumentException("无效的定时任务配置：" . get_class($task));
        }
        // 确保进程名唯一（避免冲突）
        $processName = $task->processName ?? get_class($task) . '_' . uniqid();
        new Crontab($task->cron, function () use ($task, $processName) {
            try {
                Log::info("定时任务开始执行：{$processName}");
                $task->run();
                Log::info("定时任务执行完成：{$processName}");
            } catch (Throwable $e) {
                Log::error("定时任务执行失败（{$processName}）：{$e->getMessage()}", [
                    'file' => $e->getFile(),
                    'line' => $e->getLine()
                ]);
            }
        }, $processName);
    }

    /**
     * 订阅火币行情数据并处理
     * 优化点：修复变量作用域，增强错误处理，完善重连机制
     */
    public static function currency(array $serverConfig, Worker $worker): void
    {
        if (empty($serverConfig['ws_host'])) {
            throw new InvalidArgumentException("缺少ws_host配置，无法建立连接");
        }

        // 初始化异步连接（SSL传输）
        $connection = new AsyncTcpConnection($serverConfig['ws_host']);
        $connection->transport = 'ssl';
        $lastMsgTime = time(); // 最后一次收到消息的时间（用局部变量+引用传递，避免作用域问题）

        // 连接成功回调：发送订阅请求
        $connection->onConnect = function (AsyncTcpConnection $conn) use (&$lastMsgTime) {
            try {
                $lastMsgTime = time();
                $subParam = CurrencyKline::sub();
                if (empty($subParam)) {
                    throw new InvalidArgumentException("订阅参数为空，无法订阅");
                }
                $payload = json_encode([
                    'sub' => $subParam,
                    'id' => 'tick_' . uniqid() // 生成唯一ID，避免重复
                ]);
                $conn->send($payload);
                Log::info("火币行情订阅请求已发送");
            } catch (Throwable $e) {
                Log::error("连接成功后订阅失败：{$e->getMessage()}");
                $conn->close(); // 订阅失败关闭连接，触发重连
            }
        };

        // 5分钟无消息则重启Worker（心跳检测）
        $heartbeatTimer = Timer::add(2, function () use ($worker, &$lastMsgTime) {
            if (time() - $lastMsgTime > 300) {
                Log::error("5分钟未收到火币消息，重启Worker（ID:{$worker->id}）");
                Worker::stopAll(); // 停止所有进程，由主进程重启
            }
        });

        // 每5秒检查币种数量是否变更，变更则重启
        $coinCheckTimer = Timer::add(5, function () use ($worker) {
            try {
                $currentCoins = CurrencyKline::getMarketList();
                $latestCoins = CurrencyKline::getMarketList(); // 每次获取最新列表（修复原代码变量过时问题）
                if (count($currentCoins) !== count($latestCoins)) {
                    Log::info("币种数量变更（旧:".count($currentCoins).",新:".count($latestCoins).")，重启Worker（ID:{$worker->id}）");
                    Worker::stopAll();
                }
            } catch (Throwable $e) {
                Log::error("币种数量检查失败：{$e->getMessage()}");
            }
        });

        // 接收消息回调：处理行情数据和心跳
        $connection->onMessage = function (AsyncTcpConnection $conn, string $msg) use (&$lastMsgTime) {
            try {
                $lastMsgTime = time(); // 更新最后消息时间

                // 解压消息（处理可能的解压失败）
                $data = gzdecode($msg);
                if ($data === false) {
                    throw new InvalidArgumentException("消息解压失败，原始数据：" . substr($msg, 0, 100));
                }

                // 解析JSON（处理格式错误）
                $res = json_decode($data, true);
                if (json_last_error() !== JSON_ERROR_NONE) {
                    throw new InvalidArgumentException("JSON解析失败：" . json_last_error_msg() . "，原始数据：{$data}");
                }

                // 处理心跳响应
                if (isset($res['ping'])) {
                    $conn->send(json_encode(['pong' => $res['ping']]));
                    return;
                }

                // 处理订阅成功通知
                if (!empty($res['subbed'])) {
                    Log::info("行情订阅成功", ['subbed' => $res['subbed']]);
                    return;
                }

                // 处理K线数据（1秒级）
                if (!empty($res['tick']) && !empty($res['ch'])) {
                    $code = self::getCurrencyCode($res['ch']); // 封装币种提取逻辑
                    if (empty($code)) {
                        Log::warning("无法从频道提取币种，频道：{$res['ch']}");
                        return;
                    }
                    // 存储到Redis（补充币种信息）
                    $redisKey = sprintf(self::REDIS_KLINE_1SEC_PREFIX, $code);
                    $data = $res["tick"];
                    $data["code"] = $code;
                    Redis::set($redisKey, serialize($data));
                    return;
                }

                // 其他未知消息
                Log::info("收到未知类型消息", $res);

            } catch (Throwable $e) {
                Log::error("消息处理失败：{$e->getMessage()}", [
                    'file' => $e->getFile(),
                    'line' => $e->getLine(),
                    "trace"=>$e->getTrace(),
                ]);
            }
        };

        // 连接关闭回调：自动重连
        $connection->onClose = function (AsyncTcpConnection $conn) {
            Log::warning("火币连接已关闭，3秒后尝试重连");
            $conn->reConnect(3); // 3秒后重连
        };

        // 错误回调：记录错误信息
        $connection->onError = function (AsyncTcpConnection $conn, int $code, string $msg) {
            Log::error("火币连接错误", [
                'code' => $code,
                'msg' => $msg,
                'remoteAddr' => $conn->getRemoteAddress()
            ]);
        };

        //  Worker停止时清理资源
        $worker->onWorkerStop = function () use ($connection, $heartbeatTimer, $coinCheckTimer) {
            $connection->close();
            Timer::del($heartbeatTimer);
            Timer::del($coinCheckTimer);
            Log::info("Worker停止，已清理火币连接资源");
        };

        // 启动连接
        $connection->connect();
    }

    /**
     * 组装行情推送数据
     * 优化点：修复变量未定义，增强数据校验，减少无效操作
     */
    public static function getData(array $subscribedCodes): array
    {
        $pushData = [];
        if (empty($subscribedCodes)) {
            Log::info("无订阅币种，无需组装数据");
            return $pushData;
        }

        try {
            // 生成Redis键列表（1秒K线）
            $kline1SecKeys = array_map(function ($code) {
                return sprintf(self::REDIS_KLINE_1SEC_PREFIX, strtolower($code));
            }, $subscribedCodes);

            // 批量获取1秒K线数据
            $klineDataList = Redis::getInstance()->mget($kline1SecKeys);
            if ($klineDataList === false) {
                throw new InvalidArgumentException("Redis批量获取失败：" . implode(',', $kline1SecKeys));
            }
            $list =[];
            foreach ($klineDataList as $kline) {
                $kline = unserialize($kline);
                $list[$kline["code"]] = $kline;
            }
            // 遍历处理每个币种
            foreach ($subscribedCodes as $code) {
                $codeLower = strtolower($code);
                $klineData =  $list[$code] ?? null;

                // 跳过空数据
                if (empty($klineData)) {
                    Log::warning("1秒K线数据为空，币种：{$code}");
                    continue;
                }

                // 反序列化1秒K线数据

                if (!$klineData) {
                    Log::error("1秒K线反序列化失败，币种：{$code}，数据：{$klineData}");
                    continue;
                }

                // 获取昨日K线数据
                $yesterdayKey = sprintf(self::REDIS_KLINE_YESTERDAY_PREFIX, $codeLower);
                $yesterdayData = Redis::get($yesterdayKey);
                if (empty($yesterdayData)) {
                    Log::warning("昨日K线数据为空，币种：{$code}，键：{$yesterdayKey}");
                    continue;
                }

                // 反序列化昨日K线数据
                $yesterdayKline = unserialize($yesterdayData);
                if ($yesterdayKline === false) {
                    Log::error("昨日K线反序列化失败，币种：{$code}，数据：{$yesterdayData}");
                    continue;
                }

                // 计算涨跌幅（处理数值异常）
                $yesterdayClose = floatval($yesterdayKline['close'] ?? 0);
                $currentClose = floatval($klineData['close'] ?? 0);
                if ($yesterdayClose <= 0) {
                    Log::warning("昨日收盘价无效（<=0），币种：{$code}，值：{$yesterdayClose}");
                    $change = 0;
                } else {
                    $change = 100 * (($currentClose - $yesterdayClose) / $yesterdayClose);
                }

                // 组装推送数据
                $pushData[$code] = [
                    'close' => $currentClose,
                    'change' => round($change, 4), // 保留2位小数
                    'open' => floatval($kline['open'] ?? 0),
                    'code' => $codeLower
                ];
            }
        } catch (Throwable $e) {
            Log::error("行情数据组装失败：{$e->getMessage()}", [
                'file' => $e->getFile(),
                'line' => $e->getLine(),
                'trace'=>$e->getTrace()
            ]);
        }

        return $pushData;
    }

    /**
     * 从频道名提取币种代码（封装重复逻辑）
     */
    private static function getCurrencyCode(string $channel): ?string
    {
        // 假设频道格式如：market.btcusdt.kline.1sec
        if (preg_match('/market\.([a-z0-9]+)usdt\.kline/i', $channel, $matches)) {
            return strtolower($matches[1]);
        }
        return null;
    }
}